package com.atguigu.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * ClassName: DwdTrafficBaseLogSplit
 * Package: com.atguigu.edu.realtime.app.dwd.log
 * Description:
 *
 * @Author Mr.2
 * @Create 2023/9/7 16:31
 * @Version 1.0
 */
public class DwdTrafficBaseLogSplit {
    public static void main(String[] args) {
        // TODO 1. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置


        // TODO 3. From kafka read data
        String topic = "topic_log";
        // 测试 消费者组
        String groupId = "dwd_traffic_log_split_group";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");

        // TODO 4. 对流中的数据，进行类型转换(JSONString -> JSONObject), 将脏数据 放到 侧输出流

        OutputTag<String> dirtyDataTag = new OutputTag<String>("dirtyData") {};

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        } catch (Exception e) {
                            ctx.output(dirtyDataTag, jsonStr);
                        }
                    }
                }
        );
        // For test output -> yes
//        jsonObjDS.print(">>>>");

        // TODO 5. 将 侧输出流 写到Kafka主题(topic_dirty_data)
        // 先 捕获侧输出流
        SideOutputDataStream<String> dirtyDataDS = jsonObjDS.getSideOutput(dirtyDataTag);
        // 测试脏数据 -> ok
//        dirtyDataDS.print("dirty->");
        // 写入kafka topic: topic_dirty_data
        dirtyDataDS.sinkTo(MyKafkaUtil.getKafkaSink("topic_dirty_data"));

        // TODO 6. 按照设备的id "common"-"mid"，进行分组
        KeyedStream<JSONObject, String> keyedJsonObjDS = jsonObjDS.keyBy(
                new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject jsonObj) throws Exception {
                        return jsonObj.getJSONObject("common").getString("mid");
                    }
                }
        );

        // TODO: 7. 使用 Flink的状态编程，对新老访客标记(is_new)进行修复
        SingleOutputStreamOperator<JSONObject> fixedDS = keyedJsonObjDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    // 存放上次访问的日期
                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor
                                = new ValueStateDescriptor<>("lastVisitDateState", String.class);
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {

                        String lastVisitDate = lastVisitDateState.value();

                        Long ts = jsonObj.getLong("ts");
                        String currentVisitDate = DateFormatUtil.toDate(ts);

                        // 获取 新老访客标记
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");

                        if ("1".equals(isNew)) {  // is_new = 1
                            if (StringUtils.isEmpty(lastVisitDate)) {
                                // 如果, 键控状态为null，认为本次是该访客首次访问 APP，将日志中 ts 对应的日期更新到状态中，
                                // 不对 is_new 字段做修改；
                                lastVisitDateState.update(currentVisitDate);
                            } else {
                                // 如果, 键控状态不为null，且 首次访问日期不是当日，说明访问的是老访客，将 is_new 字段置为 0；
                                if (!lastVisitDate.equals(currentVisitDate)) {  // 老访客
                                    isNew = "0";
                                    jsonObj.getJSONObject("common").put("is_new", isNew);
                                }
                            }
                        } else {  // is_new = 0
                            if (StringUtils.isEmpty(lastVisitDate)) {
                                //如果, 键控状态为 null，说明访问 APP 的是老访客, 但本次是该访客的页面日志首次进入程序。
                                //当前端新老访客状态标记丢失时，日志进入程序被判定为新访客，Flink 程序就可以纠正被
                                //误判的访客状态标记，只要将状态中的日期设置为今天之前即可。本程序选择将状态更新为昨日；
                                String yesterday = DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000);
                                lastVisitDateState.update(yesterday);
                            }
                        }
                        // 向下游传递
                        out.collect(jsonObj);
                    }
                }
        );
        // For test output ->
//        fixedDS.print("fixedDS ->");

        // TODO 8. 分流
        // 播放日志 "appVideo" - 视频侧输出流
        // 错误日志 "err" -错误侧输出流,
        // 启动日志 "start" -启动侧输出流,
        // 曝光日志 "displays" -曝光侧输出流,
        // 动作日志 "actions" -动作侧输出流,
        // 页面日志 "page" -主流.

        // 侧输出流标签
        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displaysTag = new OutputTag<String>("displaysTag") {};
        OutputTag<String> actionsTag = new OutputTag<String>("actionsTag") {};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideoTag") {};

        SingleOutputStreamOperator<String> pageDS = fixedDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<String> out) throws Exception {
                        // 1. 判断是否是错误日志
                        JSONObject errJsonObj = jsonObj.getJSONObject("err");
                        if (errJsonObj != null) {
                            // 将 错误日志 输出到 侧输出流
                            ctx.output(errTag, jsonObj.toJSONString());
                            jsonObj.remove("err");
                        }

                        // 2. 判断是否是 启动日志
                        JSONObject startJsonObj = jsonObj.getJSONObject("start");
                        JSONObject appVideoJsonObj = jsonObj.getJSONObject("appVideo");
                        if (startJsonObj != null) {
                            // 将 启动日志 输出到 侧输出流
                            ctx.output(startTag, jsonObj.toJSONString());
                        } else if (appVideoJsonObj != null) {  // 3. 判断是否是 播放日志
                            // 将 播放日志 输出到 侧输出流; videoJsonObj 应该还剩: "a"
                            ctx.output(appVideoTag, jsonObj.toJSONString());
                        } else {
                            // 获取 信息 "common" "page" "ts"
                            JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                            JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                            Long ts = jsonObj.getLong("ts");

                            // 获取 时间戳 timestamp 单位:ms
                            // 4. 判断是否是 曝光行为信息 --> 这是个JSONArray
                            JSONArray displaysJsonArray = jsonObj.getJSONArray("displays");
                            // 遍历出 所有的曝光信息
                            if (displaysJsonArray != null && displaysJsonArray.size() > 0) {
                                for (int i = 0; i < displaysJsonArray.size(); i++) {
                                    // 每遍历出一条数据, 定义一个新的JSONObj, 用于封装曝光数据
                                    JSONObject newDisplayJsonObj = new JSONObject();
                                    JSONObject oldDisplayJsonObj = displaysJsonArray.getJSONObject(i);

                                    // 封装放入
                                    newDisplayJsonObj.put("common", commonJsonObj);
                                    newDisplayJsonObj.put("page", pageJsonObj);
                                    newDisplayJsonObj.put("display", oldDisplayJsonObj);
                                    newDisplayJsonObj.put("ts", ts);
                                    // 将曝光日志 放到 曝光侧输出流
                                    ctx.output(displaysTag, newDisplayJsonObj.toJSONString());
                                }
                                jsonObj.remove("displays");
                            }

                            // 5. 判断 是否是 动作日志
                            JSONArray actionsArr = jsonObj.getJSONArray("actions");
                            if (actionsArr != null && actionsArr.size() > 0) {
                                for (int i = 0; i < actionsArr.size(); i++) {
                                    // 每遍历出一条数据, 定义一个新的JSONObj, 用于封装曝光数据
                                    JSONObject newActionJsonObj = new JSONObject();
                                    JSONObject actionJsonObj = actionsArr.getJSONObject(i);

                                    newActionJsonObj.put("common", commonJsonObj);
                                    newActionJsonObj.put("page", pageJsonObj);
                                    newActionJsonObj.put("action", actionJsonObj);
                                    newActionJsonObj.put("ts", ts);
                                    // 将 动作日志 放到动作侧输出流
                                    ctx.output(actionsTag, newActionJsonObj.toJSONString());

                                }
                                jsonObj.remove("actions");
                            }

                            // 6. 将 页面日志放到主流中, 此时:jsonObj 还剩 "common"、"page"、"ts"
                            out.collect(jsonObj.toJSONString());
                        }

                    }
                }
        );
        // 捕获 各侧输出流
        SideOutputDataStream<String> errDS = pageDS.getSideOutput(errTag);
        SideOutputDataStream<String> videoDS = pageDS.getSideOutput(appVideoTag);
        SideOutputDataStream<String> startDS = pageDS.getSideOutput(startTag);
        SideOutputDataStream<String> displayDS = pageDS.getSideOutput(displaysTag);
        SideOutputDataStream<String> actionDS = pageDS.getSideOutput(actionsTag);
        // For test output -> yes
        pageDS.print("page ->");
        videoDS.print("appVideo ->");
        errDS.print("err ->");
        startDS.print("start ->");
        displayDS.print("display ->");
        actionDS.print("action ->");

        // TODO 9. 将不同的流的数据，写到kafka不同的topic中
        pageDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_page_log"));
        videoDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_video_log"));
        errDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_err_log"));
        startDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_start_log"));
        displayDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_display_log"));
        actionDS.sinkTo(MyKafkaUtil.getKafkaSink("dwd_traffic_action_log"));

        // 执行环境
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
